package spring.cloud.tasks.server;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.client.item_task_processor.Task;
import spring.cloud.tasks.common.utils.Schedules;
import spring.cloud.tasks.common.zookeeper.ZooKeeperItemTaskDataManager;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class Boss {
    @Setter
    @Getter
    public ZooKeeperItemTaskDataManager zooKeeperItemTaskDataManager;

    @Setter
    @Getter
    private Map<String, ProcessorRunTimeInfo> processorIdToProcessorRunTimeInfoMap = new ConcurrentHashMap<String, ProcessorRunTimeInfo>();

    //管理端进行调用
    public void initTaskItems() throws Exception {
        List<Task> taskList = new ArrayList<Task>();
        for (Task task : taskList) {
            this.zooKeeperItemTaskDataManager.createTaskItems(task.getTaskId());
        }
        for (Task task : taskList) {


            zooKeeperItemTaskDataManager.clearExpireProcessors(task.getTaskId(), task.getJudgeDeadInterval());
            //
            List<String> processorIdList = zooKeeperItemTaskDataManager.loadProcessorIdList(task.getTaskId());
            zooKeeperItemTaskDataManager.cleanTaskItemCurrentProcessorIdOutOfProcessorIdList(task.getTaskId(), processorIdList);
            zooKeeperItemTaskDataManager.findAndCleanExceptionTaskItemCurrentProcessor(task.getTaskId());//清理以后再进行任务的分配
            //
            zooKeeperItemTaskDataManager.assignTaskItemWithProcessorId(task.getTaskId(), processorIdList);
        }

        this.updateTaskWorkerScheduleInfoRequestNum();

    }


    //中心服务器和客户端都需要进行自己的处理
    public void updateTaskWorkerScheduleInfoRequestNum() throws Exception {
        //只能针对现有的进行处理
        Set<String> taskIdList = new TreeSet<String>();
        List<Task> list = this.zooKeeperItemTaskDataManager.getTaskList();
        for (Task task : list) {
            taskIdList.add(task.getTaskId());
        }
        for (String taskId : taskIdList) {
            processorIdToProcessorRunTimeInfoMap.putIfAbsent(taskId, new ProcessorRunTimeInfo(taskId));
            ProcessorRunTimeInfo taskProcessorRunTimeInfo = processorIdToProcessorRunTimeInfoMap.get(taskId);
            //
            Task task = this.zooKeeperItemTaskDataManager.loadTask(taskId);
            List<String> workerIdList = this.zooKeeperItemTaskDataManager.getWorkIdListSupportedForTaskScheduleInfo(taskId);
            //允许短时间内调度器不是完全均衡的,故不必追求所有的worker都完整的注册到zk
            boolean needReBuildTaskWorkerScheduleInfo = needReBuildTaskWorkerScheduleInfo = needReBuildTaskWorkerScheduleInfo(taskId, task, taskProcessorRunTimeInfo, workerIdList);
            if (needReBuildTaskWorkerScheduleInfo) {
                int[] processorCountPerWorkers = Schedules.allot(task.getTotalprocessorCount(), workerIdList.size());
                for (int i = 0; i < workerIdList.size(); i++) {
                    String workerId = workerIdList.get(i);
                    this.zooKeeperItemTaskDataManager.updateTaskWorkerScheduleInfoRequestNum(taskId, workerId, processorCountPerWorkers[i]);
                }
                task.setOnlyOneWorkerDealWithTask(workerIdList.size() == 1);//worker需要看到
                this.zooKeeperItemTaskDataManager.updateTask(task);
                //
                taskProcessorRunTimeInfo.updateVersion();
                taskProcessorRunTimeInfo.setWorkerIdList(workerIdList);
            }
        }
    }

    public boolean needReBuildTaskWorkerScheduleInfo(String taskScheduleInfoId, Task task, ProcessorRunTimeInfo processorRunTimeInfo, List<String> workerIdList) {
        if (processorRunTimeInfo.version == null) {
            return true;
        } else {
            List<String> $workerIdList = processorRunTimeInfo.workerIdList;
            return $workerIdList.size() == workerIdList.size() && $workerIdList.containsAll(workerIdList);
        }
    }
}
